package cn.doitedu.rtdw.datasync;

import cn.doitedu.rtdw.udfs.GpsToGeohashUdf;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class JOB02_地理位置维表同步HBASE {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing(5000);
        //env.getCheckpointConfig().setCheckpointStorage("hdfs://doitedu:8020/rtdw_job1_ckpt");
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);


        // 建表，映射mysql中的源表
        tenv.executeSql(
                " create table area_mysql(                            "+
                        "     lat double,                                     "+
                        "     lng double,                                     "+
                        "     province string,                                "+
                        "     city string,                                    "+
                        "     region string                                   "+
                        " ) with (                                            "+
                        "    'connector' = 'jdbc',                            "+
                        "    'url' = 'jdbc:mysql://doitedu:3306/realtimedw',  "+
                        "    'table-name' = 'gps_area',                       "+
                        "    'username' = 'root',                             "+
                        "    'password' = 'root'                              "+
                        " )                                                   "
        );

        // 建表，映射hbase中的目标表
        tenv.executeSql(
                " create table area_hbase(                            "+
                        "     geohash string,                                  "+
                        " 	  f row<province string,city string,region string>   "+
                        " ) with (                                             "+
                        "     'connector' = 'hbase-2.2',                       "+
                        "     'table-name' = 'doit42_area',                    "+
                        "     'zookeeper.quorum' = 'doitedu:2181'              "+
                        " )                                                    "
        );

        // 写一个insert...select...
        tenv.createTemporaryFunction("gps_geo", GpsToGeohashUdf.class);
        tenv.executeSql(
                " insert into area_hbase               "+
                        " SELECT                               "+
                        "    gps_geo(lat,lng) as geohash,      "+
                        "    row(province,city,region)  as f   "+
                        " from area_mysql                      "+
                        " where lat is not null and lng is not null and lat between 10 and 60 and lng between 0 and 140  "
        );




    }
}
